package com.inns.position.hbase.service.impl;

import com.inns.position.hbase.HbaseConsts;
import com.inns.position.hbase.service.IDataService;
import com.inns.position.hbase.service.IHBaseService;
import com.inns.position.kafka.bean.LocationInfo;
import com.inns.position.kafka.bean.LocationRealTimeInfo;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;

/**
 * Created by phoenix on 2018/4/21.
 */
@Service
public class DataServiceImpl implements IDataService {

    private Logger logger = LoggerFactory.getLogger(this.getClass());
    private static final String tableName = HbaseConsts.TableName_Original;
    private static final String familyColumn = HbaseConsts.FamilyColumn;

    @Resource
    private IHBaseService ihBaseService;

    @Resource
    private RedisTemplate<String, String> redisTemplate;

    @Override
    public void save(LocationInfo locationInfo) {
        insert(locationInfo);
    }

    private void insert(LocationInfo locationInfo) {
        long t1 = System.currentTimeMillis();
        String rowKey = locationInfo.getRowKey();
//        List<Cell> cells = ihBaseService.getRowKey(tableName, rowKey, null, null);
        List<Cell> cells = null;
        long t2 = System.currentTimeMillis();
        if (cells == null || cells.size() == 0) {
            // insert
            ihBaseService.putRowValueBatch(tableName, rowKey, familyColumn, locationInfo.getTimestamp(), locationInfo.getColumnValues());
            long t3 = System.currentTimeMillis();
//            updateOrInsert(locationInfo);
            long t4 = System.currentTimeMillis();
//            logger.info("get time=" + (t2 - t1) + ", save time=" + (t3 - t2) + ", update time=" + (t4 - t3));
//            updateOrInsertPoi(locationInfo);
        } else {
            // 数据重复
            logger.info("duplicate :" + locationInfo);
        }
    }

    private void updateOrInsert(LocationInfo locationInfo) {
        long t1 = System.currentTimeMillis();
        String rowKey = locationInfo.getRowKeyShort();
        String tableName = HbaseConsts.TableName_Period;
        long curTime = locationInfo.getTimestamp();
//        String prefixKey = rowKey.substring(0, rowKey.length() - 5);
//        List<List<Cell>> cells = ihBaseService.scanPrefixRowKey(tableName, prefixKey, curTime - HbaseConsts.SPLIT_TIME(), curTime);
        String oldRowKey = getOldRowKey(locationInfo);
        long t2 = System.currentTimeMillis();
        Map<String, String> columnValues = locationInfo.getColumnValues();
        if (oldRowKey == null/*cells == null || cells.size() == 0*/) {
            // insert
            columnValues.put("startTime", curTime + "");
            logger.info("rowKey_New=" + rowKey);
            ihBaseService.putRowValueBatch(tableName, rowKey, familyColumn, null, columnValues);
        } else {
            columnValues.put("endTime", curTime + "");
//            Cell cell = cells.get(0).get(0);
//            rowKey = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
            rowKey = oldRowKey;
            logger.info("rowKey_Old=" + oldRowKey);
            ihBaseService.putRowValueBatch(tableName, rowKey, familyColumn, null, columnValues);
        }
        long t3 = System.currentTimeMillis();
        logger.info("updateOrInsert get time=" + (t2 - t1) + ", save time=" + (t3 - t2));
    }

    /**
     * @param locationInfo
     * @return 如果是第一次访问或者是新的一次访问，返回null，否则返回这次访问的 起始时间key
     */
    private String getOldRowKey(LocationInfo locationInfo) {
        HashOperations<String, String, LocationRealTimeInfo> opsForHash = redisTemplate.opsForHash();
        LocationRealTimeInfo locationRealTimeInfo = opsForHash.get(locationInfo.getIMSI(), locationInfo.getBuildId());
        if (locationRealTimeInfo == null || locationRealTimeInfo.isNewAccess(locationInfo)) {
            return null;
        }
        return locationInfo.getIMSI() + "|" + locationRealTimeInfo.getLastAccessTime().getStartTime() / 1000;
    }
}
